package com.tensai.study.java.opensource.jetlinks.upgrade_plan.comparative_testing.reactor.v3_4.emitterProcessor;

import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.RandomUtil;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
import reactor.util.concurrent.Queues;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author Tensai
 */
public class SinksManyTest {

	private final Map<String, String> stringStringMap = new ConcurrentHashMap<>();
	private final Sinks.Many<String> eventSink =
			Sinks.many().multicast().onBackpressureBuffer(Queues.SMALL_BUFFER_SIZE, false);


	@Test
	public void test() {
		eventSink.asFlux().subscribe(this::printP);
		ThreadUtil.execute(this::next);
		sleep(5000);
	}

	@Test
	public void test1() {
		eventSink.asFlux().subscribe(this::printP);
		fillDataToMap();
		ThreadUtil.execute(this::next);
		accept().subscribe(this::printF);
		sleep(5000);
	}

	private Flux<String> accept() {
		return Flux.concat(Flux.fromIterable(stringStringMap.values()), eventSink.asFlux()).distinct();
	}

	/**
	 * Sinks.Many发送数据
	 */
	private void next() {
		for (int i = 0; i < 10; i++) {
			eventSink.tryEmitNext(RandomUtil.randomString(4));
			sleep(500);
		}
	}

	/**
	 * 填充Map
	 */
	private void fillDataToMap() {
		for (int i = 0; i < 10; i++) {
			stringStringMap.put(RandomUtil.randomString(4), "---" + RandomUtil.randomString(4));
		}
	}

	private void printP(String s) {
		System.out.println("p - " + s);
	}

	private void printF(String s) {
		System.out.println("F - " + s);
	}

	/**
	 * 线程休眠
	 *
	 * @param t 休眠时间
	 */
	protected void sleep(long t) {
		try {
			Thread.sleep(t);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

}